package com.song.shop.kafka;

import org.apache.kafka.clients.producer.*;

import java.util.Properties;

public class KafkaProducerDemo {

    private static Producer<String , String> producer ;
    private final static String TOPIC = "java_topic" ;

    static {
        Properties props = new Properties();
        props.put("bootstrap.servers", "111.231.232.212:9092");
        props.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer");
        props.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer");
        producer = new KafkaProducer<>(props);
    }

    public void produce(){
        for (int i = 0 ; i < 100 ; i++){
            try {
                Thread.sleep(1000);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
            String key=String.valueOf("key"+i);
            String data="hello kafka message:"+key;
            producer.send(new ProducerRecord<>(TOPIC,key,data));
            producer.send(new ProducerRecord<>(TOPIC, key, data), new Callback() {
                @Override
                public void onCompletion(RecordMetadata recordMetadata, Exception e) {
                    //todo sth
                    System.out.println("--完成---");
                }
            });
            System.out.println(data);
        }
    }

    public static void main(String[] args) {

        KafkaProducerDemo demo = new KafkaProducerDemo() ;
        demo.produce();
    }

}
